package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo5ForeachPartition {
  def main(args: Array[String]): Unit = {
    /**
     * foreach、foreachPartition
     * 都是行为算子
     * foreach
     * 需要接收一个函数f：参数类型同RDD中的泛型，返回值类型为Unit
     * 会将每一条数据依次传递个函数f进行最终的一个处理，一般用于输出打印（测试）
     *
     * foreachPartition
     * 需要接收一个函数f：参数类型是Iterator类型，返回值类型为Unit
     * 会将每个分区的数据传给Iterator并进行最终的处理，一般用于将结果数据保存到外部系统
     *
     */

    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo5ForeachPartitions")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)


    // 1、读取students数据
    val linesRDD: RDD[String] = sc.textFile("Spark/data/stu/students.txt", 4)

    println(linesRDD.getNumPartitions) // getNumPartitions不是算子，相当于是RDD的一个属性

    // 2、建立JDBC连接
    /**
     * 算子外部的代码在Driver端执行的
     * 算子内部的代码是以Task的形式发送到Executor中执行的
     * 连接是不能被序列化的，所以连接的建立需要放入算子内部
     */
    //    val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
    //    val st: PreparedStatement = conn.prepareStatement("insert into stu values(?,?,?,?,?)")

    // 3、遍历每一条数据
    // 因为不需要返回值，所以选择foreach算子进行遍历
    //    linesRDD.foreach(line => {
    //      // 连接是不能被序列化的，所以连接的建立需要放入算子内部
    //      // foreach是针对每一条数据处理一次，相当于这里会创建1000次连接 会造成性能问题
    //      val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
    //      val st: PreparedStatement = conn.prepareStatement("insert into stu values(?,?,?,?,?)")
    //
    //      val splits: Array[String] = line.split(",")
    //      val id: Int = splits(0).toInt
    //      val name: String = splits(1)
    //      val age: Int = splits(2).toInt
    //      val gender: String = splits(3)
    //      val clazz: String = splits(4)
    //      st.setInt(1, id)
    //      st.setString(2, name)
    //      st.setInt(3, age)
    //      st.setString(4, gender)
    //      st.setString(5, clazz)
    //      st.execute()
    //      st.close()
    //      conn.close()
    //    })

    // 可以适用foreachPartition代替foreach完成对MySQL数据的插入
    // 适用于在算子内部需要跟外部数据源建立连接（一般创建连接是为了写入数据）的情况
    linesRDD.foreachPartition(iter => {
      // 连接是不能被序列化的，所以连接的建立需要放入算子内部
      // foreach是针对每一条数据处理一次，相当于这里会创建1000次连接 会造成性能问题
      // 对每个分区的数据进行处理，相当于每个分区建立一次连接，因为有4个分区，所以只会创建4次连接
      // 大大降低连接的创建次数 提高性能
      val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
      val st: PreparedStatement = conn.prepareStatement("insert into stu values(?,?,?,?,?)")

      // 这里的foreach方法实际上不是RDD的算子了，这里是Iterator的foreach方法
      // 不会出现连接未被序列化的问题，当前处理的分区的数据都会共用一个连接
      iter.foreach(line => {
        val splits: Array[String] = line.split(",")
        val id: Int = splits(0).toInt
        val name: String = splits(1)
        val age: Int = splits(2).toInt
        val gender: String = splits(3)
        val clazz: String = splits(4)
        st.setInt(1, id)
        st.setString(2, name)
        st.setInt(3, age)
        st.setString(4, gender)
        st.setString(5, clazz)
        //        st.execute() // 相当于每条数据插入一次 性能也比较低
        st.addBatch()
      })
      st.executeBatch() // 采用批量插入的方式
      st.close()
      conn.close()
    })


  }

}
